package com.david.mydemo.ctl_kafka.sender;

import com.alibaba.fastjson.JSON;
import com.david.mydemo.ctl_kafka.config.KafkaConfig;
import com.david.mydemo.ctl_kafka.config.Message;
import com.david.mydemo.util.DateUtils;
import lombok.extern.slf4j.Slf4j;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;

@Component
@Slf4j
public class KafkaSender {

//    @Autowired
//    private KafkaTemplate<String, String> kafkaTemplate;

    private final KafkaTemplate<String, String> kafkaTemplate;

    public KafkaSender(KafkaTemplate<String, String> kafkaTemplate) {
        this.kafkaTemplate = kafkaTemplate;
    }


    public void sendMessage(String msg) {
        Message message = new Message(System.currentTimeMillis(), msg, DateUtils.dateTimeNow());
        log.info("【++++++++++++++++++ 生产者message ：{}】", JSON.toJSONString(message));

        //对 topic 发送消息
//        kafkaTemplate.send(KafkaConfig.TOPICS, JSON.toJSONString(message));
//        kafkaTemplate.send(KafkaConfig.TOPICS, KafkaConfig.PARTION, KafkaConfig.MESSAGE_KEY, JSON.toJSONString(message));
        kafkaTemplate.send(KafkaConfig.TOPICS, KafkaConfig.PARTION, KafkaConfig.MESSAGE_KEY, msg);
    }

}
